package com.raylu.realtime.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Properties;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2021/12/25.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
public class KafkaSinkUtil {
    private static Properties load;

    static {
        try {
            load = PropertiesUtil.load("config.properties");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static FlinkKafkaProducer<String> getKafkaSink(String topic) throws IOException {
        Properties prop = new Properties();
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, load.getProperty("base.log.app.kafka.bootstrap-server"));
        prop.setProperty(ProducerConfig.ACKS_CONFIG, load.getProperty("base.log.app.kafka.acks"));
        prop.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, load.getProperty("base.log.app.kafka.enable.idempotence"));
         FlinkKafkaProducer<String> flinkKafkaProducer = new FlinkKafkaProducer<>(topic, new SimpleStringSchema(), prop);
         return flinkKafkaProducer;
     }

     public static <T> FlinkKafkaProducer<T> getKafkaSinkBySchema(KafkaSerializationSchema<T> kafkaSerializationSchema){
         Properties prop = new Properties();
         prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, load.getProperty("base.log.app.kafka.bootstrap-server"));
         FlinkKafkaProducer<T> flinkKafkaProducer = new FlinkKafkaProducer<T>(load.getProperty("base.db.app.kafka.default-topic"), kafkaSerializationSchema, prop, FlinkKafkaProducer.Semantic.EXACTLY_ONCE,5);
         return flinkKafkaProducer;
     }


}
